package com.yanyu.space.common.log.handler;

import com.yanyu.space.common.log.message.LogMessage;
import com.yanyu.space.common.log.message.pool.LogMessagePool;
import com.yanyu.space.common.log.sender.LogSender;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author yanyu
 */
@Slf4j
public final class LogWorker extends Thread {
    private final BlockingQueue<String> logBlockQueue = new LinkedBlockingQueue<>();
    private LogMessagePool logMessagePool;
    private int identifPos;
    private LogSender logSender;

    @SneakyThrows
    public LogWorker(int identifPos, LogSender logSender, Class<LogMessagePool> clazz) {
        this.identifPos = identifPos;
        this.logMessagePool = clazz.newInstance();
        this.logSender = logSender;

        // 设置线程属性
        this.setDaemon(true);
        this.setName("log-processor-" + identifPos);
        this.start();
    }

    /**
     * 容量大小
     *
     * @return
     */
    public int capacity() {
        return this.logMessagePool.capacity();
    }

    /**
     * 追加
     *
     * @param tarceId
     * @param logMessage
     */
    public void put(String tarceId, LogMessage logMessage) {
        logMessagePool.put(tarceId, logMessage);
    }

    /**
     * 处理
     *
     * @param traceId
     */
    public void handle(String traceId) {
        logBlockQueue.add(traceId);
    }

    @Override
    public void run() {
        while (true) {
            String traceId = null;
            try {
                traceId = logBlockQueue.take();
                LogMessage logMessage = LogWorker.this.logMessagePool.get(traceId);
                if (logMessage != null) {
                    logSender.send(logMessage);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                LogWorker.this.logMessagePool.remove(traceId);
            }
        }
    }
}
